package com.example.demo.mqserver;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: 张
 * Date: 2024-08-15
 * Time: 16:30
 */

import com.example.demo.common.Consumer;
import com.example.demo.common.ConsumerEnv;
import com.example.demo.common.MqException;
import com.example.demo.mqserver.core.*;
import com.example.demo.mqserver.datacenter.DiskDataCenter;
import com.example.demo.mqserver.datacenter.MemoryDataCenter;

import java.io.IOException;
import java.io.RandomAccessFile;
import java.security.KeyStore;
import java.util.HashMap;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;


/**
 * VirtualHost这个类表示虚拟主机
 * 每个虚拟主机下面都管理着自己的交换机,队列,绑定,消息 数据
 * 同时提供API 给 上层调用
 *
 * 这里是业务逻辑的整合,整合完之后是直接给上层调用的,所以此处的异常都要进行处理
 */
public class VirtualHost {
    private String virtualHostName;
    //内存中的数据管理
    private MemoryDataCenter memoryDataCenter = new MemoryDataCenter();
    //硬盘中的数据管理
    private DiskDataCenter diskDataCenter = new DiskDataCenter();
    //实现交换机转发和BindingKey的验证
    private Router router = new Router();
    private ConsumerManager consumerManager = new ConsumerManager(this);

    //创建一个操作交换机的锁对象
    private final Object exchangeLocker = new Object();

    //队列的锁对象
    private final Object queueLocker = new Object();

    public Router getRouter() {
        return router;
    }

    public String getVirtualHostName() {
        return virtualHostName;
    }

    public MemoryDataCenter getMemoryDataCenter() {
        return memoryDataCenter;
    }

    public DiskDataCenter getDiskDataCenter() {
        return diskDataCenter;
    }

    public VirtualHost(String name){
        this.virtualHostName = name;
        //对于MemoryDataCenter,不需要额外的初始化操作
        //DiskDataCenter需要调用其的init()方法进行初始化
        diskDataCenter.init();

        //硬盘的数据还需要恢复到内存中 ---- diskDataCenter的init()中会对对应的文件是否存在进行判断以及处理,如果是通过init新建的,数据恢复也没啥东西需要进行恢复
        try {
            memoryDataCenter.recovery(diskDataCenter);
        } catch (IOException | MqException | ClassNotFoundException e) {
            e.printStackTrace();
            System.out.println("[VirtualHost] 恢复内存数据失败!");
        }
    }

    //创建交换机
    //如果交换机不存在就创建,如果存在就返回
    //创建成功就返回true,失败就返回false
    //此处为了表示交换机和虚拟主机之间的从属关系,重新约定交换机的名字 = 虚拟主机的名字 + 交换机的原本的名字
    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType, boolean durable, boolean autoDelete,
                                   Map<String,Object> arguments){
        //把交换机名字,加上虚拟主机的名字作为前缀
        exchangeName = virtualHostName + exchangeName;
        try {
            //此处纯在线程安全问题,当两个线程同时进来的时候,可能会出现同时通过创建交换机的判定
            //然后进入方法进行交换机的创建,如果此时要创建的两个交换机的内容和属性不同,必然会在后续操作出现错误
            synchronized (exchangeLocker) {
                //1.判断该交换机是否已经存在,直接通过内存查(硬盘只是负责持久化,数据已经在初始化的时候加载到了内存中了)
                Exchange existsExchange = memoryDataCenter.getExchange(exchangeName);
                if(existsExchange != null) {
                    System.out.println("[VirtualHost] 交换机已经存在! exchangeName = " + exchangeName);
                    return true;
                }
                //2.创建交换机
                Exchange exchange = new Exchange();
                exchange.setName(exchangeName);
                exchange.setType(exchangeType);
                exchange.setDurable(durable);
                exchange.setAutoDelete(autoDelete);
                exchange.setArguments(arguments);
                //3.把交换机对象写入硬盘 -- durable为true才需要进行持久化操作
                //
                if(durable) {
                    diskDataCenter.insertExchange(exchange);
                }
                //4.把交换机对象写入内存
                memoryDataCenter.insertExchange(exchange);
                System.out.println("[VirtualHost] 交换机创建完成!!!  exchangeName = " + exchangeName);
                // 上述逻辑,先写硬盘,后写内存,目的是因为硬盘更容易写失败,如果硬盘写失败了,内存就不写了
                // 如果反过来,先写内存,后写硬盘的话,如果内存写成功了,硬盘写失败了,还需要把内存中的数据给删了(比较麻烦了)
            }
            return true;
        }catch (Exception e) {
            System.out.println("[VirtualHost] 创建交换机失败!! exchangeName = " + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    //删除交换机 -- 删除操作没啥线程安全问题,顶多就是给你多删除一次,没啥副作用,但是如果是一个在创,一个在删就可能会出现问题
    public boolean exchangeDelete(String exchangeName) {
        //针对name进行转换
        exchangeName = virtualHostName + exchangeName;
        try {
           synchronized (exchangeLocker) {
               // 1.找到对应的交换机
               Exchange toDeleteExchange = memoryDataCenter.getExchange(exchangeName);
               if(toDeleteExchange == null ) {
                   throw new MqException("[VirtualHost] 交换机不存在,无法进行删除!!!");
               }
               // 2.删除硬盘上的数据
               if(toDeleteExchange.isDurable()) {
                   //它得先有持久化存储才进行删除
                   diskDataCenter.deleteExchange(exchangeName);
               }
               // 3.删除内存中的数据
               memoryDataCenter.deleteExchange(exchangeName);
               System.out.println("[VirtualHost] 交换机删除成功!! exchangeName = " + exchangeName);
           }
           return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 删除交换机失败!! exchangeName = " + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    //创建队列
    public boolean queueDeclare(String queueName,boolean durable,boolean exclusive,boolean autoDelete,
                                Map<String,Object> arguments) {
        // 更改队列名称
        queueName = virtualHostName + queueName;
        try {
            synchronized (queueLocker) {
                //1.判定队列是否存在
                MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);
                if (existsQueue != null) {
                    System.out.println("[VirtualHost] 队列已经存在!! queueName = " + queueName);
                    return true;
                }
                MSGQueue queue = new MSGQueue();
                queue.setName(queueName);
                queue.setDurable(durable);
                queue.setExclusive(exclusive);
                queue.setAutoDelete(autoDelete);
                queue.setArguments(arguments);
                //2.先在硬盘中存储
                if (durable) {
                    diskDataCenter.insertQueue(queue);
                }
                //3.在内存中进行存储
                memoryDataCenter.insertQueue(queue);
                System.out.println("[VirtualHost] 队列创建成功!! queueName = " + queueName);

            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 创建队列失败!! queueName = " + queueName);
            e.printStackTrace();
            return false;
        }
    }

    //删除队列
    public boolean queueDelete(String queueName) {
        queueName = virtualHostName + queueName;
        try {
            synchronized (queueLocker) {
                //查询当前的队列对象
                MSGQueue toDeleteQueue = memoryDataCenter.getQueue(queueName);
                if(toDeleteQueue == null) {
                    throw new MqException("[VirtualHost] 队列不存在,无法删除!! queueName = " + queueName);
                }
                //1.先查看是否有进行持久化
                if(toDeleteQueue.isDurable()) {
                    //硬盘中进行删除
                    diskDataCenter.deleteQueue(queueName);
                }
                //2.内存中进行删除操作
                memoryDataCenter.deleteQueue(queueName);
                System.out.println("[VirtualHost] 删除队列成功!! queueName = " + queueName);
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 删除队列失败 !!");
            e.printStackTrace();
            return false;
        }
    }

    //创建绑定
    public boolean queueBind(String queueName,String exchangeName,String bindingKey) {
        queueName = virtualHostName + queueName;
        exchangeName = virtualHostName + exchangeName;
        try {
            //在创建绑定的时候,由于涉及到交换机和队列的创建问题
            //所以此处就直接加两把锁
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    //判定当前的绑定是否已经存在
                    Binding existsBinding = memoryDataCenter.getBinding(exchangeName,queueName);
                    if(existsBinding != null) {
                        throw new MqException("[VirtualHost] 绑定已经存在! exchangeName = " + exchangeName + " ,queueName = " + queueName);
                    }
                    //2.验证bandingKey是否合法(bandingKey有设定的规则)
                    if(!router.checkBindingKey(bindingKey)) {
                        throw new MqException("[VirtualHost] bindingKey非法!! bindingKey = " + bindingKey);
                    }
                    //3.创建binding
                    Binding binding = new Binding();
                    binding.setBindingKey(bindingKey);
                    binding.setQueueName(queueName);
                    binding.setExchangeName(exchangeName);
                    //4.获取一下交换机和队列,如果交换机或队列不存在,则这个绑定无法创建
                    Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                    if(exchange == null) {
                        throw new MqException("[VirtualHost] 绑定的交换机不存在!! exchangeName = " + exchangeName);
                    }
                    MSGQueue queue = memoryDataCenter.getQueue(queueName);
                    if(queue == null) {
                        throw new MqException("[VirtualHost] 绑定的队列不存在!! queueName = " + queueName);
                    }
                    //5.存!
                    //在硬盘中,得这两个是持久化(在数据文件中有数据的),再进行存储绑定
                    if(queue.isDurable() && exchange.isDurable()) {
                        diskDataCenter.insertBinding(binding);
                    }
                    memoryDataCenter.insertBinding(binding);
                    System.out.println("[VirtualHost] 绑定创建成功!!! exchangeName = " + exchangeName
                            + " ,queueName = " + queueName);
                }
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 添加绑定失败!! exchangeName = " + exchangeName
                    + " ,queueName = " + queueName);
            e.printStackTrace();
            return false;
        }
    }

    //删除绑定
    public boolean queueUnBind(String queueName,String exchangeName) {
        queueName = virtualHostName + queueName;
        exchangeName = virtualHostName + exchangeName;
        try {
            //删除绑定也和创建绑定同理
            synchronized (exchangeLocker) {
                synchronized(queueLocker) {
                    //1.获取binding,查看是否存在
                    Binding binding = memoryDataCenter.getBinding(exchangeName,queueName);
                    if(binding == null) {
                        throw new MqException("[VirtualHost] 该绑定不存在!! exchangeName = " + exchangeName
                                + " ,queueName = " + queueName);
                    }
//            //2.获取对应的队列和交换机
//            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
//            if(exchange == null) {
//                throw new MqException("[VirtualHost] 对应的交换机不存在!! exchangeName = " + exchangeName);
//            }
//            MSGQueue queue = memoryDataCenter.getQueue(queueName);
//            if(queue == null) {
//                throw new MqException("[VirtualHost] 对应的队列不存在!! queueName = " + queueName);
//            }
//                    3.删
//            if(queue.isDurable() && exchange.isDurable()) {
//                diskDataCenter.deleteBinding(binding);
//            }
                    //胜在简单
                    //无论绑定是否持久化,都在硬盘尝试删一下
                    diskDataCenter.deleteBinding(binding);
                    memoryDataCenter.deleteBinding(binding);
                    System.out.println("[VirtualHost] 删除绑定成功!!");
                }
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 删除绑定失败!!");
            e.printStackTrace();
            return false;
        }
    }

    //发送消息到指定的队列中
    public boolean basicPublic(String exchangeName,String routingKey,BasicProperties basicProperties,byte[] body) {
        try {
            //1.名称的转换
            exchangeName = virtualHostName + exchangeName;
            //2.检查routingKey 是否合法
            if(!router.checkRoutingKey(routingKey)) {
                throw new MqException("[VirtualHost] routingKey非法!! routingKey = " + routingKey);
            }
            //3.找到交换机对象
            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
            if(exchange == null) {
                throw new MqException("[VirtualHost] 指定交换机不存在!! exchangeName = " + exchangeName);
            }
            //4.判定交换机类型
            if(exchange.getType() == ExchangeType.DIRECT) {
                //直接交换机
                //以 routingKey 来作为队列的名字,直接把消息写入指定的队列中   此时,可以无视绑定关系
                String queueName = virtualHostName + routingKey;
                //5.查找对应对象
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                //6.构造消息对象
                Message message = Message.createMessageWithId(routingKey,basicProperties,body);
                if(queue == null) {
                    throw new MqException("[VirtualHost] 指定队列不存在!! queueName = " + queueName);
                }
                //7.消息转发
                sendMessage(queue,message);
            } else {
                //按照 扇出交换机 和 主题交换机的形式进行转发
                //5.找到该交换机关联的所有的绑定,并遍历绑定对象
                ConcurrentHashMap<String,Binding> bindingsMap = memoryDataCenter.getBindings(exchangeName);
                //遍历
                for (Map.Entry<String,Binding> entry : bindingsMap.entrySet()) {
                    //1) 获取到绑定对象,判定对应的队列是否存在
                    Binding binding = entry.getValue();
                    MSGQueue queue = memoryDataCenter.getQueue(binding.getQueueName());
                    if(queue == null) {
                        //不希望因为一个队列的失败,影响到其他队列接受消息
                        System.out.println("[VirtualHost] basicPublic在发送消息时,发现绑定队列不存在!! queueName = " + binding.getQueueName());
                        continue;
                    }
                    //2)构造消息对象
                    Message message = Message.createMessageWithId(routingKey,basicProperties,body);
                    //3)判定该消息是否可以转发给这个队列
                    //  此处如果是扇形交换机,要向所有绑定的队列转发消息的.
                    //  此处如果是主题交换机,就要进行bindingKey 和 routingKey 是否匹配
                    if(!router.route(exchange.getType(),binding,message)) {
                        continue;
                    }
                    //转发消息给对应队列
                    sendMessage(queue,message);
                }

            }
            return true;
        }catch (Exception e) {
            System.out.println("[VirtualHost] 发送消息失败!!");
            e.printStackTrace();
            return  false;
        }
    }


    private void sendMessage(MSGQueue queue, Message message) throws IOException, MqException, InterruptedException {
        //判断消息是否要进行持久化 ---- 1为非持久化,2为持久化
        int deliverMode = message.getDeliverMode();
        //写入硬盘
        if(deliverMode == 2) {
            diskDataCenter.sendMessage(queue,message);
        }
        //写入内存
        memoryDataCenter.sendMessage(queue,message);
        //通知消费者收到消息(往阻塞队列上添加令牌) --- 通知消费者消费
        consumerManager.notifyConsume(queue.getName());
    }

    // 订阅消息 --- 添加一个队列的订阅者
    // 当队列收到消息之后,就要把消息推送给对应的订阅者
    // consumerTag 表示消费者的身份标识
    // autoAck: 消息被消费完成后,应答的方式,True:自动应答,False:消费者手动应答
    // Consumer : 是一个回调函数,此处类型设定为函数式接口,这样后续调用basicConsume并且实参的时候,就可以写作 lambda
    public boolean basicConsume(String consumerTag, String queueName, boolean autoAck, Consumer consumer) {
        // 构造一个 ConsumerEnv 对象,把这个对应的队列给找到,再把这个ConsumerEnv对象给添加到该队列中
        queueName = virtualHostName + queueName;
        try {
            consumerManager.addConsumer(consumerTag,queueName,autoAck,consumer);
            System.out.println("[VirtualHost] basicConsume成功!! queueName = " + queueName);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] basicConsume失败!! queueName = " + queueName);
            e.printStackTrace();
            return false;
        }
    }

    //消费者手动应答
    public boolean basicAck(String queueName,String messageId) {
        try {
            //1.获取到对应的消息和队列数据
            queueName = virtualHostName + queueName;
            Message message = memoryDataCenter.getMessage(messageId);
            MSGQueue queue = memoryDataCenter.getQueue(queueName);
            if(message == null) {
                throw new MqException("[VirtualHost] 要确认的消息不存在!! messageId = " + messageId);
            }
            if(queue == null) {
                throw new MqException("[VirtualHost] 要确认的队列不存在!! queueName = " + queueName);
            }
            //2.删除硬盘上的数据
            if(message.getDeliverMode() == 2) {
                diskDataCenter.deleteMessage(queue,message);
            }
            //3.删除消息中心中的数据
            memoryDataCenter.removeMessage(messageId);
            //4.删除待确认集合中的数据
            memoryDataCenter.removeMessageWaitAck(queueName,messageId);
            System.out.println("[VirtualHost] basicAck成功,消息被成功确认!!!  queueName = " + queueName +
                    " ,messageId = " + messageId);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] basicAck失败,消息确认失败!!! queueName = " + queueName +
                    " ,messageId = " + messageId);
            e.printStackTrace();
            return false;
        }
    }

}
